package com.ht.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ht.shortlink.project.common.convention.exception.ServiceException;
import com.ht.shortlink.project.dao.entity.*;
import com.ht.shortlink.project.dao.mapper.*;
import com.ht.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.ht.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.ht.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import static com.ht.shortlink.project.common.constant.RedisCacheConstant.LOCK_SHORT_LINK_GID_UPDATE_KEY;
import static com.ht.shortlink.project.common.constant.ShortLinkConstant.IP_AMAP_URL;

/**
 * 短链接监控状态保存消息队列消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {
    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkTodayStatsMapper linkTodayStatsMapper;
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;
    private final RedissonClient redissonClient;
    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        if(messageQueueIdempotentHandler.isMessageQueueProcessed(id.toString())){
            if(messageQueueIdempotentHandler.isMessageQueueAccomplished(id.toString())){
                return;
            }
            throw new ServiceException("消息未完成流程，需要重试");
        }
        try {
            Map<String, String> producerMap = message.getValue();
            String gid = producerMap.get("gid");
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            actualShortLinkStats(gid, statsRecord);
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex){
            log.error("短链接监控数据处理异常");
            messageQueueIdempotentHandler.deleteMessageQueueProcessedKey(id.toString());
        }
        messageQueueIdempotentHandler.setMessageQueueAccomplished(id.toString());
    }

    public void actualShortLinkStats(String gid, ShortLinkStatsRecordDTO statsRecord){
        String fullShortUrl = statsRecord.getFullShortUrl();
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_SHORT_LINK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        if(!rLock.tryLock()){
            delayShortLinkStatsProducer.send(statsRecord);
            return;
        }
        try{
            if(StrUtil.isBlank(gid)){
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }

            Date date = new Date();
            int hour = DateUtil.hour(date, true);
            Week week = DateUtil.dayOfWeekEnum(date);
            int weekValue = week.getIso8601Value();
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(date)
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);

            Map<String, Object> requestMap = new HashMap<>();
            requestMap.put("ip", statsRecord.getRemoteAddr());
            requestMap.put("key", statsLocaleAmapKey);
            String localeResultStr = HttpUtil.get(IP_AMAP_URL, requestMap);
            JSONObject localeResultObj = JSONUtil.parseObj(localeResultStr);
            String infoCode = localeResultObj.getStr("infocode");

            if(StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")){
                // 记录地区访问数据
                String province = localeResultObj.getStr("province");
                String city = localeResultObj.getStr("city");
                String adcode = localeResultObj.getStr("adcode");
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .province(ObjectUtil.equals(province, "[]") ? "未知" : province)
                        .city(ObjectUtil.equals(city, "[]") ? "未知" : city)
                        .adcode(ObjectUtil.equals(adcode, "[]") ? "未知" : adcode)
                        .country("中国")
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleStats(linkLocaleStatsDO);

                // 记录操作系统访问记录
                LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .os(statsRecord.getOs())
                        .build();
                linkOsStatsMapper.shortLinkOsStats(linkOsStatsDO);

                // 记录浏览器访问记录
                LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .browser(statsRecord.getBrowser())
                        .build();
                linkBrowserStatsMapper.shortLinkBrowserStats(linkBrowserStatsDO);

                // 记录设备访问记录
                LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .device(statsRecord.getDevice())
                        .build();
                linkDeviceStatsMapper.shortLinkDeviceStats(linkDeviceStatsDO);

                // 记录网络访问记录
                LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .network(statsRecord.getNetwork())
                        .build();
                linkNetworkStatsMapper.shortLinkNetworkStats(linkNetworkStatsDO);

                // 记录访问日志
                LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .user(statsRecord.getUv())
                        .network(statsRecord.getNetwork())
                        .device(statsRecord.getDevice())
                        .locale(StrUtil.join("-", linkLocaleStatsDO.getCountry(), linkLocaleStatsDO.getCity(), linkLocaleStatsDO.getProvince()))
                        .browser(statsRecord.getBrowser())
                        .os(statsRecord.getOs())
                        .ip(statsRecord.getRemoteAddr())
                        .build();
                linkAccessLogsMapper.insert(linkAccessLogsDO);

                // 记录短链接总计访问量
                shortLinkMapper.incrTotalStats(gid, fullShortUrl, 1, (statsRecord.getUvFirstFlag() ? 1 : 0), (statsRecord.getUipFirstFlag() ? 1 : 0));

                // 记录今日访问量
                LinkTodayStatsDO linkTodayStatsDO = LinkTodayStatsDO.builder()
                        .gid(gid)
                        .fullShortUrl(fullShortUrl)
                        .todayPv(1)
                        .todayUv((statsRecord.getUvFirstFlag() ? 1 : 0))
                        .todayUip((statsRecord.getUipFirstFlag() ? 1 : 0))
                        .date(date)
                        .build();
                linkTodayStatsMapper.insertLinkTodayStats(linkTodayStatsDO);
            }
        } catch (Throwable ex){
            log.error("短链接访问量统计异常", ex);
        } finally {
            rLock.unlock();
        }
    }
}
